package com.chb.flink.state

import com.chb.flink.source.StationLog
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector


/**
 * 案例需求：计算每个手机的呼叫间隔时间，单位是毫秒
 * 第一种方法
 */
object TestKeyState2 {
    def main(args: Array[String]): Unit = {
        val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
        import org.apache.flink.streaming.api.scala._

        //读取文件数据
        val data = streamEnv.readTextFile(getClass.getResource("/station.log").getPath)
            .map(line => {
                var arr = line.split(",")
                new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
            })


        data.keyBy(_.callOut)
            .flatMapWithState[(String, Long), StationLog] {
                case (in: StationLog, None) => (List((in.callOut, 0)), Some(in)) // 第一次呼叫
                case (in: StationLog, pre: Some[StationLog]) => { // 再次会叫，状态的更新，以及时间间隔的计算
                    val interval = in.callTime - pre.get.callTime
                    (List((in.callOut, interval)), Some(in))
                }
            }
                .print()

        streamEnv.execute()

    }

}
